package cgl.sensorgrid.sopac.gps.filters;

import cgl.narada.event.NBEvent;
import cgl.sensorgrid.filters.Filter;
import java.io.*;
import java.nio.*;
import java.nio.charset.*;
import cgl.narada.service.time.TimeService;
import cgl.narada.service.time.ntptime.TimeServiceImpl;


public class MultiRYOPublisher extends Filter implements Runnable {

    private String ryoFilePath;
    private long publishFrequency;
    private int msgcount=0;
    private static TimeService t;
    private String threadname=""; 

    public MultiRYOPublisher(String[] args, String[] argsb) {
        if (args == null || args.length == 0) {
            System.err.println(
                    "Incorrect number of arguments in ryo2ascii::Constructor...");
        } else if (argsb.length == 3) {

            //init the NTP service
            t = TimeServiceImpl.getInstance();

            //Sleep for 5 seconds after initializing the NTP service
            try {
                System.out.println(
                        "Sleeping for 3 seconds before initializing the NTP service");
                Thread.sleep(3000);
            } catch (InterruptedException ex1) {
                ex1.printStackTrace();
            }

            if (argsb[0] != null && argsb[0].length() > 0)
                this.ryoFilePath = (argsb[0]);

            //How many messages to be published in one second
            if (argsb[1] != null && argsb[1].length() > 0)
                this.publishFrequency = 1000 / Long.parseLong(argsb[1]);

            if (argsb[2] != null && argsb[2].length() > 0)
                this.threadname = (argsb[2]);
            
            
            this.initialize(args);
            //publishRYOFile(ryoFilePath);

        }
    }

    public void publishRYOFile(String filePath) {
        String line = "";
        try {

            //Finding EOF markers inside the RYO File to determine the end
            //of individual network messages.

            DataInputStream dis = new DataInputStream(new FileInputStream(
                    filePath));

            int start = 0;
            int markerIndex = 0;
            byte[] by = new byte[2048];
            byte[] message = new byte[2048];

            Charset charset = Charset.forName("ISO-8859-1");
            CharsetDecoder decoder = charset.newDecoder();
            CharSequence eofMarker = new String("[EOF]");

            //System.out.println("Remaining bytes to be published: ");

            while (dis.available() > 0) {

                //System.out.print(dis.available()+"\t");

                if (dis.available() > 2048)
                    by = new byte[2048];
                else
                    by = new byte[dis.available()];

                dis.read(by);

                ByteBuffer bbuf = ByteBuffer.wrap(by);

                CharBuffer cbuf = decoder.decode(bbuf);
                String s = cbuf.toString();

                if (s.contains(eofMarker)) {
                    //markerIndex is the first index of EOF marker from the beginning of the dis,
                    //hence the lengh of the byte array to hold this particular GPS message

                    markerIndex = s.indexOf("[EOF]");
                    message = new byte[markerIndex];

                    //copy the message from the current dis cursor to markerIndex.
                    System.arraycopy(by, 0, message, 0, markerIndex);

                    //Create the event property stamp
                    msgcount++;
//                    String msgstamp =  msgcount + "\t" + message.length + "\t" + t.getTimestamp();
                    String msgstamp =  msgcount + "\t" + message.length + "\t" + System.currentTimeMillis();
                    if(msgcount%100==1)
                    {
                    	System.out.println("msgcount: "+msgcount + " | threadname:"+threadname);
                    }
                    //Here publish the data
                    this.publishDataWithProperty(message, "MSGSTAMP", msgstamp);

                    //System.out.println("Sleeping for " + this.publishFrequency + " milisecs..");
                    Thread.sleep(this.publishFrequency);

                    //System.out.println("markerIndex  = " + markerIndex );

                    //move the cursor to the new message
                    start = start + markerIndex + 6;
                    //System.out.println("start = " + start);

                    //close this dis
                    dis.close();

                    //open a new dis and skip the already published messages
                    dis = new DataInputStream(new FileInputStream(filePath));
                    dis.skipBytes(start);
                }
            }
        } catch (Exception ex) {
            System.out.println(line);
            ex.printStackTrace();
        }
    }

	/**
	 * 
	 * @param event
	 *            NBEvent
	 */
	public void run() {
		try {
	           publishRYOFile(ryoFilePath);
			return;
		} catch (Exception ex) {
			ex.printStackTrace();
		}
	}
    
    public void onEvent(NBEvent nbEvent) {
    }

    public static void main(String args[]) {
        if (args.length < 5) {
            System.out.println("Usage: java cgl.sensorgrid.sopac.gps.filters.RYOPublisher commType nbHost nbPort pubTopic ryoFilePath publishFrequency");
            System.exit(0);
        }

        String commType = args[0];
        String nbHost = args[1];
        String nbPort = args[2];
        String pubTopic = args[3];
        String ryoFilePath = args[4];
        String publishFrequency = args[5];
        long IncreasingQuantity=1;
        long SpaceOfTime=1800;
        long TotalQuantity=1;
        long CurThreadNumber=0;
        
        if (args[6] != null && args[6].length() > 0)
        	IncreasingQuantity=Long.parseLong(args[6]);

        if (args[7] != null && args[7].length() > 0)
        	SpaceOfTime=Long.parseLong(args[7]);

        if (args[8] != null && args[8].length() > 0)
        	TotalQuantity=Long.parseLong(args[8]);        
        
//        String commType = "tcp";
//        String nbHost = "gf3.ucs.indiana.edu";
//        String nbPort = "5544";
//        String pubTopic = "/SOPAC/GPS/CRTN_01/RYO1";
//        String ryoFilePath =
//                "C:/home/gaydin/sensorgrid/data/ryo/CRTN_01-09_10_2006-05_05_17_PM.ryo";
//        String publishFrequency = "1";
        while(CurThreadNumber<TotalQuantity)
        {
            int ii=1;
            while (ii<=IncreasingQuantity)
            {
            	CurThreadNumber=CurThreadNumber+1;
            	String ThreadName = ""+CurThreadNumber;
                String[] argsa = {commType,
                                 nbHost,
                                 nbPort,
                                 "",
                                 pubTopic+CurThreadNumber};

                String[] argsb = {ryoFilePath, 
                		         publishFrequency,
                                 ThreadName
                                 };        	
                new Thread(new MultiRYOPublisher(argsa, argsb)).start();
                ii=ii+1;
                try {
            		Thread.sleep(5000);
            	} catch (InterruptedException e) {
            		// TODO Auto-generated catch block
            		e.printStackTrace();
            	}            
            }
            for(int jj=1;jj<=SpaceOfTime;jj++)
            {
            	try {
					Thread.sleep(1000);
				} catch (InterruptedException e) {
					// TODO Auto-generated catch block
					e.printStackTrace();
				}
            }
            
        }
        System.out.println("done !");
//        String ThreadName = "1";
//        String[] argsa = {commType,
//                         nbHost,
//                         nbPort,
//                         "",
//                         pubTopic};
//
//        String[] argsb = {ryoFilePath, 
//        		         publishFrequency,
//                         ThreadName
//                         };
        //MultiRYOPublisher publish = new MultiRYOPublisher(argsa, argsb);
        //publish.run();
        
//        new Thread(new MultiRYOPublisher(argsa, argsb), ""+1).start();
// 
//        ThreadName = "2";
//        String[] argsc = {ryoFilePath, 
//		         publishFrequency,
//                ThreadName
//                };
//
//        new Thread(new MultiRYOPublisher(argsa, argsc),""+2).start();

        //ThreadName = "2";
        //MultiRYOPublisher publish1 = new MultiRYOPublisher(argsa, argsb);
        //publish1.run();
    }
}
